exports.__esModule = true;

var _asyncMqtt = require('async-mqtt');

var _asyncMqtt2 = _interopRequireDefault(_asyncMqtt);

function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }

function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }

// const defaultIp = think.config('mqttServer');

/**
 * Mqtt采集服务基类
 */
class MqttClientService {
    constructor(mqttServerIp) {
        this.serverIp = mqttServerIp || think.config('mqttServer');
        this.client = _asyncMqtt2.default.connect(this.serverIp, {
            clientId: 'mqtt_' + Math.random().toString(16).substr(2, 8),
            clean: true,
            resubscribe: true
        });
        this.onConnect(() => {
            this.connectStatus = true; //连接状态
        });
        this.onErro(() => {
            // this.client.end();
            this.connectStatus = false;
            think.logger.error('error');
        });

        this.onReconnect(() => {
            think.logger.warn('Mqtt is reconnecting');
        });
        this.onClose(() => {
            this.connectStatus = false;
            think.logger.warn('mqtt is disconnect');
        });
    }

    /**
     * 订阅主题
     * @param topic  String/Array 主题
     * @param qos  number  消息传输等级，取值0-2
     * @returns {promise}
     */
    subscribe(topic, qos = 1) {
        var _this = this;

        return _asyncToGenerator(function* () {
            try {
                return yield _this.client.subscribe(topic, { qos: qos });
            } catch (e) {
                think.logger.error(new Error(e.message));
                return null;
            }
        })();
    }

    /**
     * 向主题发布消息
     * @param topic   String/Array发布的主题名称
     * @param message 发布的消息体
     * @returns {*}
     */
    publish(topic, message) {
        var _this2 = this;

        return _asyncToGenerator(function* () {
            try {
                return yield _this2.client.publish(topic, message);
            } catch (e) {
                think.logger.error(new Error(e.message));
                return null;
            }
        })();
    }

    /**
     * 取消订阅
     * @param topic  String/Array 主题名称
     * @returns {*}
     */
    unsubscribe(topic) {
        var _this3 = this;

        return _asyncToGenerator(function* () {
            try {
                return yield _this3.client.unsubscribe(topic);
            } catch (e) {
                think.logger.error(new Error(e.message));
                return null;
            }
        })();
    }

    /**
     * 接收数据,在sub订阅以后，调用该函数
     * @param callback(主题名称,数据)
     */
    getMessage(callback) {
        this.client.on('message', (topic, message) => {
            let jsonData = think.bufferToJsonAndUtf8(message); //转码
            callback(topic, jsonData);
        });
    }

    /**
     * 连接事件
     */
    onConnect(callback) {
        this.client.on('connect', callback);
    }

    /**
     * 连接错误事件
     */
    onErro(callback) {
        this.client.on('error', callback);
    }

    /**
     * 重连事件
     * @param callback
     */
    onReconnect(callback) {
        this.client.on('reconnect', callback);
    }

    /**
     *关闭事件
     */
    onClose(callback) {
        this.client.on('close', callback);
    }

    /**
     * 自定义事件函数
     */
    on(event, callback) {
        this.client.on(event, callback);
    }

    /**
     * 关闭连接
     */
    end() {
        this.client.end();
    }

}
exports.default = MqttClientService;